package com.galeno.javabasic;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import static org.apache.flink.table.api.Expressions.$;

/**
 * @author galeno
 * @Title:
 * @Description: use kafka to mysql
 * @date 2021/10/3020:59
 */
public class TestDebeziumFlink {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",12345);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        // topic created by subscription
        String topicName = "test-topic";
        // replace with cdc endpoint
        String bootstrapServer = "galeno01:9092";
        // consumer group id
        String groupID = "test-flink";

        // create table connected to source
        tableEnvironment.executeSql("CREATE TABLE test_table (\n" +
                "    c1 INT,\n" +
                "    `f:col1` STRING,\n" +
                "    `f:col2` STRING,\n" +
                "    `f:col3` STRING\n" +
                ") WITH (\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic'     = '"+ topicName + "',\n" +
                "    'properties.bootstrap.servers' = '" + bootstrapServer + "',\n" +
                "    'properties.group.id' = '" + groupID + "',\n" +
                "    'debezium-json.schema-include' = 'true',\n" +
                "    'format'    = 'debezium-json'\n" +
                ")");

        // create print table
        tableEnvironment.executeSql("CREATE TABLE test_table_copy (\n" +
                "    c1 INT,\n" +
                "    `f:col1` STRING,\n" +
                "    `f:col2` STRING,\n" +
                "    `f:col3` STRING\n" +
                ") WITH (\n" +
                "    'connector' = 'print'" +
                ")");

        Table testTable = tableEnvironment.from("test_table");
        testTable.select($("c1"), $("f:col1"), $("f:col2"), $("f:col3")).executeInsert("test_table_copy");

        tableEnvironment.executeSql("SELECT * FROM test_table").print();
    }
}